package com.spider.moudle.pool;

import com.google.common.collect.Maps;
import com.spider.moudle.api.Writer;
import com.spider.moudle.entity.WriterType;
import com.spider.moudle.util.PropertiesUtil;
import com.spider.moudle.writer.FileWriter;
import com.spider.moudle.writer.KafkaWriter;

import java.util.LinkedList;
import java.util.Map;

/**
 * Created by Linbo Hu on 2018/1/15.
 *
 */
public class WriterPool {

    private static WriterPool instance = new WriterPool();

    public static WriterPool getInstance() {
        return instance;
    }

    private final Map<String, LinkedList<Writer>> poolMap = Maps.newConcurrentMap();

    private WriterPool() {
        Integer size = Integer.valueOf(PropertiesUtil.getVal("main.kafka.writer.pool.size"));
        LinkedList<Writer> fileWriterList = new LinkedList<>();
        for(int i = 0; i < size; i++){
            FileWriter writer = new FileWriter();
            fileWriterList.add(writer);
        }
        synchronized(poolMap) {
            poolMap.put(WriterType.FILE.toString(), fileWriterList);
            poolMap.notifyAll();
        }
        LinkedList<Writer> kafkaWriterList = new LinkedList<>();
        for(int i = 0; i < size; i++){
            //KafkaWriter writer = new KafkaWriter();
            //kafkaWriterList.add(writer);
        }
        synchronized(poolMap) {
            poolMap.put(WriterType.KAFKA.toString(), kafkaWriterList);
            poolMap.notifyAll();
        }

    }


    /**
     * 释放一个crawler
     * @param writerType
     * @return
     * @throws InterruptedException
     */
    public Writer fetch(WriterType writerType) throws Exception {
        LinkedList<Writer> pool = poolMap.get(writerType.toString());
        if (pool == null) {
            throw new Exception("该pool不存在");
        }
        synchronized (pool) {
            while (pool.isEmpty()) {
                pool.wait();
            }
            return pool.removeFirst();
        }
    }

    /**
     * 释放一个crawler
     * @param writerType
     * @param writer
     */
    public void release(WriterType writerType, Writer writer) {
        LinkedList<Writer> pool = poolMap.get(writerType.toString());
        if(pool == null) {
            System.err.println(writerType.toString() + " 下不存在对应的pool");
        } else {
            if (writer != null) {
                synchronized (pool) {
                    pool.addLast(writer);
                    pool.notifyAll();
                }
            }
        }
    }
}
